package com.dcloud.mq.listen;


import com.dcloud.mq.annotation.RocketMqMsgListener;
import com.dcloud.mq.config.ConsumerConfig;
import com.dcloud.mq.config.RocketConfig;
import com.dcloud.mq.consumer.SimpleConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author shijf
 * @Description 根据注解中的参数自动创建消费者
 * @Date 10:45 2021/11/24
 **/
@Slf4j
@Component
public class MqMsgLoadListener implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {
    //扫描的类
    private Map<String, Object> beanMap = new HashMap<>();
    //消费者列表
    private List<SimpleConsumer> consumerList = new ArrayList<>();

    @Value("${rocketmq.name-server}")
    private String address;

    public void onApplicationEvent(ContextRefreshedEvent event) {
        //如果是第二次加载则不处理了
        if(event.getApplicationContext().getParent() != null) {
            return;
        }

        //获取注解类
        beanMap = event.getApplicationContext().getBeansWithAnnotation(RocketMqMsgListener.class);
        log.info("Consumers 正在创建---------------------------------------");
        log.info("共有"+beanMap.size()+"个消费者----------------------------");
        //未扫描到不处理
        if(beanMap == null || beanMap.size() == 0) {
            return;
        }
        for(Object bean : beanMap.values()) {
            createConsumer(bean);
        }
    }

    /**
     * 创建消费者
     * @param bean
     */
    private void createConsumer(Object bean) {
        if(!(bean instanceof MessageListenerConcurrently)) {
            return;
        }
        //获取注解
        RocketMqMsgListener mqMsgListener = bean.getClass().getAnnotation(RocketMqMsgListener.class);

        //配置
        ConsumerConfig consumerConfig = new ConsumerConfig();

        consumerConfig.setTopic(mqMsgListener.topic());
        consumerConfig.setConsumerGroup(mqMsgListener.consumerGroup());
        consumerConfig.setSelectorExpression(mqMsgListener.selectorExpression());
        consumerConfig.setMessageModel(mqMsgListener.messageModel());
        if(!StringUtils.isEmpty(mqMsgListener.instanceName())) {
            consumerConfig.setInstanceName(mqMsgListener.instanceName());
        }
        RocketConfig rocketConfig = new RocketConfig(address);

        //创建消费者
        SimpleConsumer consumer = new SimpleConsumer(rocketConfig, consumerConfig, (MessageListenerConcurrently)bean);
        consumerList.add(consumer);
        //初始化并启动
        try {
            consumer.init();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 服务停止
     */
    @Override
    public void destroy() throws Exception {
        for(SimpleConsumer consumer : consumerList) {
            if(consumer == null) {
                continue;
            }
            consumer.destroy();
        }
    }
}
